package com.cti.commons.mqtt.handle;

import javax.annotation.Resource;

import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.StringUtils;

import com.cti.commons.mqtt.constant.MsgReveiceStatus;
import com.cti.commons.mqtt.repository.MqttReceiveMessageLog;
import com.cti.commons.mqtt.repository.MqttReceiveMessageRepository;
import com.cti.commons.mqtt.thread.MessageContext;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 抽象消费支持类
 */
@Slf4j
public abstract class AbstractSuperConsumer implements IMqttMessageListener {

    public static final int SIZE = 1024;

    @Value("${consumer.max.failTimes:5}")
    private Integer consumerMaxFailTimes;

    @Resource(name = "mqttConsumeExecutor")
    private ThreadPoolTaskExecutor executor;

    @Autowired
    private MqttReceiveMessageRepository repository;


    @Override
    public void messageArrived(String topic, MqttMessage message) {
        MqttReceiveMessageLog mqttReceiveMessageLog = new MqttReceiveMessageLog(message.getId() + "", topic,
            message.toString(),
            null,
            message.getQos());
        try {
            repository.save(mqttReceiveMessageLog);
            //放入上下文对象，待拦截器拦截@MqttTransaction注解后，需要等业务处理完再更新状态
            MessageContext.setReceiveMessageLog(mqttReceiveMessageLog);
            executor.submit(() -> {
                SpringUtil.getBean(this.getClass()).consumerHandler(message.toString());
            });
        } catch (Exception e) {
            log.error("执行本地业务发生异常", e);
            String errorMessageString = e.getMessage();
            if (!StringUtils.isEmpty(errorMessageString)) {
                mqttReceiveMessageLog.setException(errorMessageString.length() < SIZE ? errorMessageString :
                    errorMessageString.substring(0, SIZE));
            }
            mqttReceiveMessageLog.setFailTimes(mqttReceiveMessageLog.getFailTimes() + 1);
            if (mqttReceiveMessageLog.getFailTimes() >= this.consumerMaxFailTimes) {
                mqttReceiveMessageLog.setStatus(MsgReveiceStatus.MAX_FAIL.getStatus());
            } else {
                mqttReceiveMessageLog.setStatus(MsgReveiceStatus.FAIL.getStatus());
            }
            repository.save(mqttReceiveMessageLog);
        } finally {
            MessageContext.removeReceiveMessageLog();
        }

    }

    protected abstract void consumerHandler(String payload);

}
